class Auth:
    """
    Base class for all authentication schemes.
    To implement a custom authentication scheme, subclass `Auth` and override
    the `.auth_flow()` method.
    If the authentication scheme does I/O, such as disk access or network calls, or uses
    synchronization primitives such as locks, you should override `.async_auth_flow()`
    to provide an async-friendly implementation that will be used by the `AsyncClient`.
    Usage of sync I/O within an async codebase would block the event loop, and could
    cause performance issues.
    """

    requires_request_body = False
    requires_response_body = False

    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
        """
        Execute the authentication flow.
        To dispatch a request, `yield` it:
        ```
        yield request
        ```
        The client will `.send()` the response back into the flow generator. You can
        access it like so:
        ```
        response = yield request
        ```
        A `return` (or reaching the end of the generator) will result in the
        client returning the last response obtained from the server.
        You can dispatch as many requests as is necessary.
        """
        # We could *potentially* do something more clever here to return
        # more helpful errors if `sync_auth_flow`/`async_auth_flow` *has* been overridden,
        # but the auth flow is being used in the incorrect sync/async context.
        raise NotImplementedError("Override 'auth_flow' to implement a custom auth class.")
        yield request  # pragma: nocover 

    def sync_auth_flow(
        self, request: Request
    ) -> typing.AsyncGenerator[Request, Response]:
        """
        Execute the authentication flow synchronously.

        By default, this defers to `.auth_flow()`. You should override this method
        when the authentication scheme does I/O, such as disk access or network calls,
        or uses concurrency primitives such as locks.
        """
        if self.requires_request_body:
            request.read()

        flow = self.auth_flow(request)
        request = next(flow)

        while True:
            response = yield request
            if self.requires_response_body:
                response.read()

            try:
                request = flow.send(response)
            except StopIteration:
                break

    async def async_auth_flow(
        self, request: Request
    ) -> typing.AsyncGenerator[Request, Response]:
        """
        Execute the authentication flow asynchronously.

        By default, this defers to `.auth_flow()`. You should override this method
        when the authentication scheme does I/O, such as disk access or network calls,
        or uses concurrency primitives such as locks.
        """
        if self.requires_request_body:
            await request.aread()

        flow = self.auth_flow(request)
        request = next(flow)

        while True:
            response = yield request
            if self.requires_response_body:
                await response.aread()

            try:
                request = flow.send(response)
            except StopIteration:
                break

auth = httpx.BasicAuth()
request = httpx.Request("GET", "https://www.example.com")

# The initial request should include a basic auth header.
flow = auth.sync_auth_flow(request)
request = next(flow)
assert request.headers['Authorization'].startswith('Basic')

# No other requests are made.
response = httpx.Response(text='Hello, world!', status_code=200)
with pytest.raises(StopIteration):
    flow.send(response)

auth = httpx.DigestAuth()
request = httpx.Request("GET", "https://www.example.com")

# The initial request should not include an auth header.
flow = auth.sync_auth_flow(request)
request = next(flow)
assert 'Authorization' not in request.headers

# If a 200 response is returned, then no other requests are made.
response = httpx.Response(text='Hello, world!', status_code=200)
with pytest.raises(StopIteration):
    flow.send(response)

auth = httpx.DigestAuth()
request = httpx.Request("GET", "https://www.example.com")

# The initial request should not include an auth header.
flow = auth.sync_auth_flow(request)
request = next(flow)
assert 'Authorization' not in request.headers

# If a 401 response is returned, then a digest auth request is made.
headers = {'WWW-Authenticate': 'Digest realm="...", qop="...", nonce="...", opaque="..."'}
response = httpx.Response(text='Auth required', status_code=401, headers=headers)
request = flow.send(response)
assert request.headers['Authorization'].startswith('Digest')

# No other requests are made.
response = httpx.Response(text='Hello, world!', status_code=200)
with pytest.raises(StopIteration):
    flow.send(response)